package ETLLOG2;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class LoggerMapper extends Mapper<LongWritable,Text,Text,NullWritable>{

    Text k = new Text();
    @Override
    protected void map(LongWritable key,
                       Text value,
                       Context context) throws IOException, InterruptedException {
        //1.Text ===> String
        String line = value.toString();

        //2.解析日志
        LogBean bean = parseLog(line);

        //3.判断非法数据
        if (!bean.isValid()) {
            return;
        }

        k.set(bean.toString());

        //4.输出
        context.write(k,NullWritable.get());
    }

    /**
     * 解析日志
     *
     * @param line
     * @return
     */
    private LogBean parseLog(String line) {
        LogBean logBean = new LogBean();

        //1.截取后的字段属性
        String[] fields = line.split(" ");
        //筛选条件1：字段长度大于11
        if (fields.length > 11) {
            //2.封装数据
            logBean.setRemote_addr(fields[0]);
            logBean.setRemote_user(fields[1]);
            logBean.setTime_local(fields[3].substring(1));
            logBean.setRequest(fields[6]);
            logBean.setStatus(fields[8]);
            logBean.setBody_bytes_sent(fields[9]);
            logBean.setHttp_referer(fields[10]);

            //如果字段长度大于12，就拼接浏览器来源
            //筛选条件2：如果有浏览器来源就拼接上，看是否大于12
            if (fields.length > 12) {
                logBean.setHttp_user_agent(fields[11] + " " + fields[12]);
            }else {
                //浏览的相关信息
                logBean.setHttp_user_agent(fields[11]);
            }

            //判断状态码
            //筛选条件3：状态码大于等于400，是非法数据
            if (Integer.parseInt(logBean.getStatus()) >= 400) {
                logBean.setValid(false);
            }
        }else {
            logBean.setValid(false);
        }
        return logBean;
    }


}
